package com.spx.chapter06;

import com.spx.chapter05.pojo.Event;
import com.spx.util.SampleDataUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-02  0:00
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class WatermarkTest {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Flink1.12 之后默认的时间语义就是EventTime，因此需要再指定
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<Event> dataSource1 = env.fromCollection(SampleDataUtil.getSample())
                /*
                 *水位线的指定要尽可能的早，距离source越近越好，因为watermark是会向下游传播的，因此设置的越早对下游来说越精确
                 * 使用Flink内部已经实现的 watermarkStrategy
                 * 使用有序流的 watermarkStrategy
                 * */
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        // 指定watermarkStrategy
                        .<Event>forMonotonousTimestamps()
                        // 指定时间戳如何提取
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                );



        SingleOutputStreamOperator<Event> dataSource2 = env.fromCollection(SampleDataUtil.getSample())
                // 指定乱序流的 watermarkStrategy,设置乱序程度
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                );
    }


}
